package com.shallowUniverse.dmp.service;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.shallowUniverse.dmp.entity.Params;
import com.shallowUniverse.dmp.mapper.TransferMapper;
import com.shallowUniverse.dmp.pojo.*;
import com.shallowUniverse.dmp.pojo.vo.TransferVo;
import com.shallowUniverse.dmp.utils.DataSourceUtils;
import com.shallowUniverse.dmp.utils.SidGenerator;
import com.shallowUniverse.dmp.utils.SqlUtils;
import org.apache.commons.dbutils.QueryRunner;
import org.apache.commons.dbutils.handlers.MapHandler;
import org.apache.commons.dbutils.handlers.MapListHandler;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;
import org.springframework.util.ObjectUtils;

import javax.annotation.Resource;
import java.sql.SQLException;
import java.util.*;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;

/**
 * @author Shuai Guo
 * @version 1.0
 * @description
 * @date 2021/8/5
 */
@Service
public class TransferService extends ServiceImpl<TransferMapper, Transfer> {

    @Resource
    private DataBaseService dataBaseService;
    @Resource
    private TableService tableService;
    @Resource
    private TableFieldService tableFieldService;
    @Resource
    private TransferLogsService transferLogsService;
    @Resource
    private TransferMapper transferMapper;
    @Resource
    private TransferFieldService transferFieldService;
    @Resource
    private SidGenerator sidGenerator;

    @Async
    @Transactional
    public void transfer(Long transferId) {
        //根据id拿到转化对象
        Transfer transfer = getById(transferId);
        //查询来源和目标
        DataBase fromDataBase = dataBaseService.getById(transfer.getFromDatabase());
        DataBase toDataBase = dataBaseService.getById(transfer.getToDatabase());
        //查询来源表和目标表
        Table fromTable = tableService.getById(transfer.getFromTable());
        Table toTable = tableService.getById(transfer.getToTable());
        //查询转换字段
        LambdaQueryWrapper<TransferField> transferFieldLambdaQueryWrapper = new LambdaQueryWrapper<>();
        transferFieldLambdaQueryWrapper.eq(TransferField::getTransferId, transferId);
        List<TransferField> transferFields = transferFieldService.list();
        //查询来源表的字段信息
        List<TableField> fromFields = tableFieldService.listByIds(transferFields.stream().map(TransferField::getFromField).collect(Collectors.toList()));
        //查询目标表的字段信息
        List<TableField> toFields = tableFieldService.listByIds(transferFields.stream().map(TransferField::getToField).collect(Collectors.toList()));

        //生成来源和目标操作对象
        QueryRunner fromRunner = new QueryRunner(DataSourceUtils.createDataSource(fromDataBase));
        QueryRunner toRunner = new QueryRunner(DataSourceUtils.createDataSource(toDataBase));

        //定义源数据集合
        List<Map<String, Object>> fromValues = new ArrayList<>();
        //定义目标数据集合
        List<Map<String, Object>> toValues = new ArrayList<>();
        //定义页码
        AtomicInteger page = new AtomicInteger(1);
        //构造修改语句
        String updateSql = SqlUtils.createUpdate(toFields, toTable, transfer);
        //定义转换总量、添加总量、更新总量
        AtomicLong total = new AtomicLong(0);
        AtomicLong insertCount = new AtomicLong(0);
        AtomicLong updateCount = new AtomicLong(0);
        //保存转换日志
        TransferLogs transferLogs = new TransferLogs();
        transferLogs.setSid(sidGenerator.generate());
        transferLogs.setStatus(1);
        transferLogs.setStartTime(new Date());
        transferLogs.setTransferId(transferId);
        transferLogsService.save(transferLogs);
        //修改transfer状态为运行
        transfer.setStatus(3);
        transferMapper.updateById(transfer);

        //开始处理
        do {
            //构造添加语句
            StringBuilder insertSql = SqlUtils.createInsert(toFields, toTable);
            //定义新增记录数变量
            AtomicInteger insertCounts = new AtomicInteger(0);

            Params<Map<String, Object>> params = new Params<>();
            params.setSize(1000);
            params.setCurrent(page.get());

            //构造源数据查询语句
            String fromQuerySql = SqlUtils.createQuery(fromDataBase, fromTable, fromFields, transfer.getFromSyncField(), params);
            //构造目标表的查询语句
            String toQuerySql = SqlUtils.createQuery(toDataBase, toTable, toFields, transfer.getToSyncField(), params);
            //查询数据
            try {
                fromValues = fromRunner.query(fromQuerySql, new MapListHandler());
                toValues = toRunner.query(toQuerySql, new MapListHandler());
            } catch (SQLException th) {
                th.printStackTrace();
            }

            //判断查出的数据是否为空
            if (!CollectionUtils.isEmpty(fromValues)) {
                //累加总量
                total.getAndAdd(fromValues.size());
                //定义新增修改参数值集合
                List<Object[]> insertParams = new ArrayList<>();
                List<Object[]> updateParams = new ArrayList<>();

                //循环处理
                List<Map<String, Object>> toValuesCopy = toValues;
                fromValues.forEach(value -> {
                    AtomicBoolean b = new AtomicBoolean(false);
                    //遍历目标集合判断是否有该数据
                    toValuesCopy.forEach(toValue -> {
                        if (!ObjectUtils.isEmpty(value.get(transfer.getFromSyncField())) && !ObjectUtils.isEmpty(toValue.get(transfer.getToSyncField()))
                                && value.get(transfer.getFromSyncField()).toString().equals(toValue.get(transfer.getToSyncField()).toString())) {
                            b.set(true);
                        }
                    });
                    //判断是否存在
                    if (!b.get()) {
                        insertSql.append(SqlUtils.createInsertValuesBatch(toDataBase, fromFields, toFields, value));
                        insertCounts.incrementAndGet();
                    } else {
                        //构造修改值
                        Object[] param = SqlUtils.createUpdateValues(toDataBase, toFields, transfer, value);
                        updateParams.add(param);
                    }
                });

                try {
                    //判断是否需要新增
                    if (insertCounts.get() > 0) {
                        toRunner.execute(insertSql.substring(0, insertSql.length() - 1));
                    }
                    //执行更新
                    int[] updateCounts = toRunner.batch(updateSql, SqlUtils.createBatchValues(updateParams));
                    //累加
                    insertCount.getAndAdd(insertCounts.get());
                    updateCount.getAndAdd(Arrays.stream(updateCounts).sum());
                } catch (SQLException throwable) {
                    throwable.printStackTrace();
                }
            }

            page.getAndIncrement();

        } while (!CollectionUtils.isEmpty(fromValues));

        //更新日志对象
        transferLogs.setStatus(2);
        transferLogs.setTotalCount(total.get());
        transferLogs.setInsertCount(insertCount.get());
        transferLogs.setUpdateCount(updateCount.get());
        transferLogs.setEndTime(new Date());
        //更新
        transferLogsService.updateById(transferLogs);

        //更新transfer状态为暂停
        transfer.setStatus(1);
        transferMapper.updateById(transfer);
    }

    @Transactional
    public void save(TransferVo transferVo) {
        //设置sid
        transferVo.getTransfer().setSid(sidGenerator.generate());
        //添加转换对象
        transferMapper.insert(transferVo.getTransfer());
        //设置转换字段的sid
        transferVo.getTransferFields().forEach(transferField -> transferField.setSid(sidGenerator.generate()));
        //保存转换字段
        transferFieldService.saveBatch(transferVo.getTransferFields());
    }
}
